package top.suven.base.core.kafka;

import top.suven.base.util.utilize.JsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 这是kafka 接入spring boot 管理的新实现方式用例;旧模式已去用了
 */

//@Component
public class KafkaConsumerListener {

	private static Logger logger = LoggerFactory.getLogger(KafkaConsumerListener.class);

//	@Autowired
    KafkaConsumerThreadPool kafkaThreadPool;

//    @KafkaListener(topics = {"top-test"})
    public void testListener(ConsumerRecord record) {
        logger.warn(" ConsumerRecord messager this is the top-test send message = : " + JsonUtils.toJsonString(record));
        kafkaThreadPool.run(new ConsumerWorkerThread(record,new KafkaConsumerHandler()));
    }
//
//    @KafkaListener(topics = {"top-test-2"})
    public void testListener2(ConsumerRecord<String,String> record) {
        logger.warn(" ConsumerRecord testListener2 this is the top-test-2 send message =-------- : " + record.value().toString());
        kafkaThreadPool.run(new ConsumerWorkerThread(record,new KafkaConsumerHandler()));
    }

//    @Component
    public class KafkaConsumerHandler extends KafkaConsumerAbstractHandler{

        public void executeConsumer(Object value ){
            logger.warn(" abstract  executeConsumer value:[{}]" , JsonUtils.toJsonString(value));
        }

    }


}

